package BanyuanClub.Study.Day0310.生产者消费者模式;

import java.util.concurrent.*;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

public class MyArrayBlockingQueue2 <T>{
    private Object[] arr;
    private int size;
    private ReentrantLock lock = new ReentrantLock();
    private Condition notFull = lock.newCondition();
    private Condition notEmpty = lock.newCondition();

    public MyArrayBlockingQueue2(int limit){
        this.arr = new Object[limit];
    }

    public void put(T obj) throws InterruptedException {
        //线程安全的map
//        ConcurrentHashMap map;
        lock.lock();
        try {
            while (size == arr.length) {
                System.out.println("队列已满，生产者等待中，释放锁");
                //让当前线程等待，并且释放锁
                notFull.await();
            }
            arr[size++] = obj;
            //不会死锁 唤醒所有的等待的线程
            notEmpty.signalAll();
            System.out.println("生产者放入了一个对象");
        }finally {
            lock.unlock();
        }
    }

    public synchronized  T take() throws InterruptedException {
        lock.lock();
        try {
            while (size == 0) {
                System.out.println("队列已空，消费者等待中，释放锁");
                notEmpty.await();
            }
            Object obj = arr[size - 1];
            arr[--size] = null;
            System.out.println("消费者取出了一个对象");
            notFull.signalAll();
            return (T) obj;
        } finally {
            lock.unlock();
        }
    }

    public static void main(String[] args) {
        MyArrayBlockingQueue2<String> q = new MyArrayBlockingQueue2<>(5);
        ExecutorService pool = new ThreadPoolExecutor(2, 2,
                0, TimeUnit.SECONDS, new SynchronousQueue<>());

        //生产者线程
        pool.execute(() -> {
            while (true) {
                try {
                    Thread.sleep(1000);
                    q.put("aaa");
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        });

        //消费者线程
        pool.execute(() -> {
            while (true) {
                try {
                    Thread.sleep(5000);
                    q.take();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        });
    }
}
